package com.shujia.kafka

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

import scala.io.Source

object Demo1Produce {
  def main(args: Array[String]): Unit = {

    /**
      * 生产者
      *
      */


    val properties: Properties = new Properties

    //咨询参数配置
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    //key value序列化类
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")


    val prducer: KafkaProducer[String, String] = new KafkaProducer[String, String](properties)


    Source
      .fromFile("spark/data/students.txt")
      .getLines()
      .foreach(line => {

        val clazz: String = line.split(",")(4)


        //判断班级属于哪一个分区   hash分区
        val partiton: Integer = Math.abs(clazz.hashCode) % 3


        //构建一个消息
        val record: ProducerRecord[String, String] =
          new ProducerRecord[String, String]("student1", partiton, null, line)


        //发送消息
        prducer.send(record)

      })




    //关闭连接
    prducer.close()


  }
}
